/*********************************************************************************
 *      Copyright:  (C) 2022 xiezongyi<1764417007@qq.com>
 *                  All rights reserved.
 *
 *       Filename:  thread_worker.c
 *    Description:  This file 
 *                 
 *        Version:  1.0.0(13/05/22)
 *         Author:  xiezongyi <1764417007@qq.com>
 *      ChangeLog:  1, Release initial version on "13/05/22 15:50:05"
 *                 
 ********************************************************************************/

#include <time.h>
#include <stdio.h>
#include <errno.h>
#include <string.h>
#include <pthread.h>

#include "logger.h"
#include "database.h"
#include "temperature.h"
#include "cJSON_worker.h"
#include "mosquitto.h"
#include "mosquitto_conf.h"
#include "mosquitto_callback_sub.h"

#define	KEEPALIVE	60
#define	LEN			256

extern int				g_stop;

void get_sys_time(char *datatime);
int  get_sample_data(sample_data_t *s_data, char *sample);

//sub线程工作函数
void *thread_sub(void *args)
{
	int                     rv = -1;
	struct mosquitto		*mosq = NULL;
	mqtt_ctx_t              *mqtt = (mqtt_ctx_t *)args;

	//传参合法性检测
	if(!args)
	{
		printf("%s() get invalid arguments\n", __FUNCTION__);
		pthread_exit(NULL);
	}

	//创建新的客户端实例
	mosq = mosquitto_new(mqtt->clientid, true, mosq);
	if(!mosq)
	{
		printf("mosquitto_new() faliure: %s\n", strerror(errno));
		goto cleanup;
	}
	printf("sub create new mosquitto OK!\n");

	//配置客户端的用户名和密码
	rv = mosquitto_username_pw_set(mosq, mqtt->username, mqtt->passwd);
	if(MOSQ_ERR_SUCCESS != rv)
	{
		printf("mosquitto_username_pw_set() failure: %s\n", strerror(errno));
		goto cleanup;
	}
    printf("%s\n", mqtt->hostname);
    printf("%d\n", mqtt->port);
    printf("%s\n", mqtt->pubtopic);
    printf("%s\n", mqtt->subtopic);
    printf("%s\n", mqtt->clientid);
    printf("%s\n", mqtt->passwd);
    printf("%s\n", mqtt->username);

	//配置回调函数
	mosquitto_connect_callback_set(mosq, on_connect_sub);
	mosquitto_disconnect_callback_set(mosq, on_disconnect_sub);
	mosquitto_subscribe_callback_set(mosq, on_subscribe);
	mosquitto_message_callback_set(mosq, on_message_sub);

	//连接服务器
	rv = mosquitto_connect(mosq, mqtt->hostname, mqtt->port, KEEPALIVE);
	if(MOSQ_ERR_SUCCESS != rv)
	{
		printf("mosquitto_connect() failure: %s\n", strerror(errno));
		goto cleanup;
	}

	//异步循环（非阻塞函数，一旦disconnect必须调用stop关闭线程）
	rv = mosquitto_loop_start(mosq);
	if(MOSQ_ERR_SUCCESS != rv)
	{
		printf("mosquitto_loop_start() failure: %s\n", strerror(errno));
		goto cleanup;
	}

	while(!g_stop)
	{
		//阻塞函数（false:等待线程结束，必须disconnect，此时该函数才会返回）
		mosquitto_loop_stop(mosq, false);
		pthread_exit(NULL);
	}

cleanup:
	mosquitto_destroy(mosq);
	mosquitto_lib_cleanup();
	pthread_exit(NULL);
}

//线程thread_pub工作函数
void *thread_pub(void *args)
{
	int                     rv = -1;
	int                    	sample_flag = 0;
	int                    	connect_flag = 0;
	char                   	datatime[64];
	char                  	sample[64];
	char                	data[64];
	char                  	buf[256];
	float                 	temper;
	time_t            		now_time = 0;
	time_t                	last_time = 0;
	sample_data_t			s_data;
	mqtt_ctx_t            	*mqtt = (mqtt_ctx_t *)args;
	struct  mosquitto       *mosq = NULL;

	if(!args)
	{
		printf("%s() get invalid arguments\n", __FUNCTION__);
		pthread_exit(NULL);
	}

	//创建新的客户端实例
	mosq = mosquitto_new(mqtt->clientid, true, mosq);
	if(!mosq)
	{
		printf("mosquitto_new() faliure: %s\n", strerror(errno));
		goto cleanup;
	}
	printf("pub create new mosquitto OK!\n");

	//配置客户端的用户名和密码
	rv = mosquitto_username_pw_set(mosq, mqtt->username, mqtt->passwd);
	if(MOSQ_ERR_SUCCESS != rv)
	{
		printf("mosquitto_username_pw_set() failure: %s\n", strerror(errno));
		goto cleanup;
	}
	printf("%s\n", mqtt->hostname);
	printf("%d\n", mqtt->port);
	printf("%s\n", mqtt->pubtopic);
	printf("%s\n", mqtt->subtopic);
	printf("%s\n", mqtt->clientid);
	printf("%s\n", mqtt->passwd);
	printf("%s\n", mqtt->username);

	while(!g_stop)
	{
		now_time = time(NULL);
		if(now_time - last_time >= mqtt->limit_time)
		{
			//获取采样的数据
			if( get_sample_data(&s_data, sample) < 0 )
			{
				printf("Get the sample failure: %s\n", strerror(errno));
			}

			//打包数据成JSON格式
			if(packet_json(mqtt->buf_json, &s_data) < 0 ) 
			{
				printf("Packet data failure: %s\n", strerror);
			}

			last_time = now_time;
			sample_flag = 1;
		}

		//判断是否连接上服务器(connect_flag=0,断开)
		if(0 == connect_flag)
		{
			rv = mosquitto_connect(mosq, mqtt->hostname, mqtt->port, KEEPALIVE);
			//连接失败
			if(MOSQ_ERR_SUCCESS != rv)
			{
				if(sample_flag)
				{
					//将此时获取的数据插入数据库
					rv = database_insert(mqtt->db, sample);
					if(rv < 0)
					{
						printf("Save data to database failure: %s\n", strerror(errno));
						goto cleanup;
					}
					else
					{
						printf("Save data to database OK!\n");
					}
				}
				connect_flag = 0;
				continue;
			}

			//连接成功
			else
			{
				rv = mosquitto_loop_start(mosq);
				if(MOSQ_ERR_SUCCESS != rv)
				{
					printf("mosquitto_loop_start() failure: %s\n", strerror(errno));
					goto cleanup;
				}
				else
				{
					connect_flag = 1;//连接成功
				}
			}
		}

		//publisher连接上server，开始发布数据
		if(connect_flag)
		{
			if(sample_flag)//如果采样成功
			{
				rv = mosquitto_publish(mosq, NULL, mqtt->pubtopic, strlen(mqtt->buf_json)+1, mqtt->buf_json, 0, true);
				if(MOSQ_ERR_SUCCESS != rv)
				{
					printf("publish message failure: %s\n", strerror(errno));

					//发布失败，关闭创建的进程
					if(mosquitto_loop_stop(mosq, true) != MOSQ_ERR_SUCCESS)
					{
						printf("Mosquitto loop stop failure: %s\n",strerror(errno));
					}

					//发布数据失败，存入数据库
					rv = database_insert(mqtt->db, sample);
					if(rv < 0)
					{
						printf("Save data to database failure: %s\n", strerror(errno));
						goto cleanup;
					}
					printf("Insert data to database OK!\n");
					connect_flag = 0;//连接标志设为断开
				}
				else
				{
					printf("Publish the latest data OK!\n");
				}
			}

			//若成功取出数据库中第一条数据，证明数据库不为空
			if(0 == database_pop_first(mqtt->db, data))
			{
				if(database_get_first(mqtt->db, &s_data) < 0)
				{
					printf("Get first data and fill the struct failure: %s\n", strerror(errno));
				}

				if(packet_json(mqtt->buff, &s_data) < 0 )
				{
					printf("Packet data faliure: %s\n", strerror(errno));
				}

				//发布从数据库中取出的数据
				rv = mosquitto_publish(mosq, NULL, mqtt->pubtopic, strlen(mqtt->buff)+1, mqtt->buff, 0, true);
				if(MOSQ_ERR_SUCCESS != rv)
				{
					printf("Publish message to server failure: %s\n", strerror(errno));

					//发布失败，关闭创建的进程
					if(mosquitto_loop_stop(mosq, true) != MOSQ_ERR_SUCCESS)
					{
						printf("Mosquitto loop stop failure: %s\n",strerror(errno));
					}
					connect_flag = 0;
					goto cleanup;
				}
				else
				{
					printf("Publish data from database OK!\n");

					//发布数据成功就将其删除
					rv = database_delete_first(mqtt->db);
					if(rv < 0)
					{
						printf("Delete data from database failure: %s\n", strerror(errno));
					}
					else
					{
						printf("Delete data from database OK!\n");
					}
				}
			}

			//数据库为空
			else
			{
				printf("The database is NULL!\n");
			}
		}
	}

cleanup:
	printf("program will exit\n");
	mosquitto_destroy(mosq);
	mosquitto_lib_cleanup();
	pthread_exit(NULL);
}

//获取系统时间
void get_sys_time(char *datatime)
{
    time_t      tmp;
    struct tm   *timp;

    time(&tmp);
    timp = localtime(&tmp);

    memset(datatime, 0, sizeof(datatime));
    strftime(datatime, 128, "%Y-%m-%d %H:%M:%S", timp);

    return;
}

//获取采样的数据
int get_sample_data(sample_data_t *s_data, char *sample)
{
	if( get_temperature(&(s_data->temp)) < 0 )
	{
		log_error("Get the ds18b20 temperature failure: %s\n", strerror(errno));
		return -1;
	}

	strncpy(s_data->ID, "xiezongyi001", sizeof(s_data->ID));
	get_sys_time(s_data->datatime);

	memset(sample, 0, sizeof(sample));
	snprintf(sample, LEN, "%s/%s/%.1f", s_data->ID, s_data->datatime, s_data->temp);

	return 0;
}

